package cn.org.queue.manage;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * @Package: cn.org.queue
 * @ClassName: ConnectionManager
 * @Description:
 * @Author: mac-pro
 * @CreateDate: 2019/2/14 上午10:36
 * @Version: 1.0
 */
@Slf4j
public class RabbitQueueManager {
    private static final String HOST = "localhost";
    private static final String USER = "admin";
    private static final String PASS = "admin";
    private static final int PORT = 5672;//默认端口
    private static final String VIRTUAL_HOST = "/";
    private static final String URI = String.format("amqp://%s:%s@%s",USER,PASS,HOST);
    protected static final String QUEUE_TYPE = "x-delayed-message";//支持消息延迟
    protected static final String EXCHANGE_NAME = "master_exchange";
    protected Connection connection;
    protected Channel channel;
    public RabbitQueueManager() { init(true); }

    /**
     * 初始化
     * 如果想消息持久化需要具备三个条件
     * exchange 持久化 queue持久化 message持久化
     * @param isDefault
     */
    public void init(boolean isDefault) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setVirtualHost(VIRTUAL_HOST);
        try {
            if (isDefault) {
                factory.setUri(URI);
            } else {
                factory.setHost(HOST);
                factory.setUsername(USER);
                factory.setPassword(PASS);
                factory.setPort(PORT);
            }
            connection = factory.newConnection();
            channel = connection.createChannel();
            Map<String, Object> args = new HashMap<>();
            // 声明x-delayed-type类型的exchange 支持消息延时
            args.put("x-delayed-type", "topic");
            /**
             * 声明转发器
             * 参数1：交换机名称
             * 参数2：交换机类型
             * 参数3：交换机持久性，如果为true则服务器重启时不会丢失
             * 参数4：交换机在不被使用时是否删除
             * 参数5：交换机的其他属性
             * */
            channel.exchangeDeclare(EXCHANGE_NAME, QUEUE_TYPE, false, false, args);
        } catch (Exception e) {
            log.error("init rabbit queue is error", e);
        }
    }

    /**
     * 设置queue参数并绑定
     */
    protected void setQueueType(String queueName, String routingKey) {
        try {
            //声明一个队列 - 持久化
            channel.queueDeclare(queueName, true, false, false, null);
            //设置通道预取计数
            channel.basicQos(1);
            //将消息队列绑定到Exchange
            channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
        } catch (IOException e) {
            log.error("bind queue is error", e);
        }
    }

    /**
     * 关闭通道
     */
    public void closed(){
        try {
            this.channel.close();
            this.connection.close();
        } catch (Exception e) {
            log.error("closed is error",e);
        }
    }
}
